package brokerFailureSuperController;

import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.util.Enumeration;
import java.util.HashMap;
import java.util.Properties;
import java.util.Set;

import javax.jms.Message;
import javax.jms.MessageListener;
import javax.jms.TopicSubscriber;

import org.apache.qpid.client.TopicPublisherAdapter;
import org.apache.qpid.client.message.JMSTextMessage;

import qpidClient.QpidClient;

/**
 * Creates all the elements in the broker failure detection system.
 * Assigns a broker to new clients that are created.
 * Reassigns a broker to clients who are using a broker that fails.
 */
public class SuperController extends QpidClient implements MessageListener {
	
    //topic used to receive messages from publishers
	protected final String LISTEN_TOPIC = "brokerRequests";
	//root of the topic used to send messages to publishers
	//(the publisher ID must be added)
	protected final String PUB_ROOT_TOPIC = "brokerUpdates.";
	//topic used to receive messages from controller
	protected final String TOPIC_SUPERCONTROLLER
			= "brokerFailure.superController";
	
	//IP address of the controller
	protected String controllerIP;
	//number of brokers in the system
	protected int numBrokers;
	//list of IP addresses of the machines in which brokers are running and the
	//ports in which they are listening
	protected HashMap<String, Integer> brokers
			= new HashMap<String, Integer>();
	
	//properties of the system
	private Properties props;
	//name of the file in which the properties are saved
	private final String PROPS_FILE_NAME = "properties.xml";
	//description of the file
	private final String PROPS_FILE_DESC = "This file contains the data that "
			+ "allows elements in the system to know where to find other "
			+ "elements in the system.";
	
	/**
	 * Constructor.
	 * Reads information about the network from PROPS_FILE_NAME, creates the
	 * elements in the broker failure detection system and starts listening to
	 * messages from the controller or new clients.
	 */
	public SuperController() {
		
		super(); //assign ID
		
		//read properties from properties file
		try {
			FileInputStream fis = new FileInputStream(PROPS_FILE_NAME);
			props = new Properties();
			props.loadFromXML(fis);
			controllerIP = props.getProperty("controllerIP");
			numBrokers = Integer.parseInt(props.getProperty("numBrokers"));
			for (int i = 1; i <= numBrokers; i++) {
				String ip = "broker" + i + "IP";
				String port = "broker" + i + "port";
				String brokerIP = props.getProperty(ip);
				int brokerPort = Integer.parseInt(props.getProperty(port));
				brokers.put(brokerIP, brokerPort);
			}
			System.out.println("Controller IP: " + controllerIP);
			System.out.println("Number of brokers: " + numBrokers);
			System.out.println("Brokers IPs: " + brokers);
		} catch (Exception e) {
			e.printStackTrace();
		}
		
		try {
			//create and start a connection to the controller broker and a
			//session in it
			openConnAndSess(controllerIP);
		} catch (Exception e) {
			e.printStackTrace();
		}

		subscribe(LISTEN_TOPIC);
		subscribe(TOPIC_SUPERCONTROLLER);
		
		//create controller
		new ControllerCreatorThread(controllerIP, brokers);
		
		Set<String> brokerAddr = brokers.keySet();
		
		//create detectors
		for (String broker : brokerAddr) {
			new DetectorCreatorThread(broker, controllerIP, brokers);
		}
	}
	
	/**
	 * Subscribe to the given topic.
	 * 
	 * @param topicName  name of the topic to subscribe to
	 */
	public void subscribe(String topicName) {
		try {
			//create a subscriber for the topic
			TopicSubscriber subscriber = getSubscriberForTopic(topicName);
			//Set its message listener
			subscriber.setMessageListener(this);
			
			System.out.println("Super controller waiting for messages in "
							   + topicName);
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	/**
	 * Invoked when a message arrives.
	 * Analyses the message:
	 * If it is from a client it is a broker request. A broker is assigned to
	 * the client and a message is sent to the client.
	 * If it is from the controller, it is a broker failure. A new broker is
	 * assigned to all clients who were using the failed broker and a message
	 * is sent to them.
	 * 
	 * @param msg  message received
	 */
	@Override
	public void onMessage(Message msg) {
				
		try {
			//get client id from the message
			JMSTextMessage message = (JMSTextMessage) msg;
			String text = message.getText();
			String[] parts = text.split(":");
						
			if (parts[0].equals("broker request")) {
				String pub_id = parts[1]; //id of the client
				//assign a broker to the client
				//TODO: they shouldn't be always the same
				String assignedBrk = "localhost";
				String assignedPort = "5674";
				props.setProperty(pub_id, assignedBrk);
				FileOutputStream fos = new FileOutputStream(PROPS_FILE_NAME);
				props.storeToXML(fos, PROPS_FILE_DESC);
				//create name of the topic to which the client is listening
				String topicName = PUB_ROOT_TOPIC + pub_id;
				//create publisher for topic
				TopicPublisherAdapter publisher =
						getPublisherForTopic(topicName);
				//create text of the message
				String assignation = assignedBrk + ":" + assignedPort;
				//send the message to the publisher
				sendMessage(publisher, assignation);
				System.out.println("Supercontroller assigned broker "
								   + assignedBrk + " and port " + assignedPort
								   + " to client " + pub_id);
			} else if (parts[0].equals("broker failed")) {
				String brokerFailed = parts[1];
				//get the list of publishers affected
				Enumeration<String> propNames =
								   (Enumeration<String>) props.propertyNames();
				while (propNames.hasMoreElements()) {
					String propName = propNames.nextElement();
					if (!propName.startsWith("broker") &&
						!propName.startsWith("controller") &&
						((String)props.get(propName)).equals(brokerFailed)) {
						String pub_id = propName;
						//assign a broker to the publisher
						//TODO: they shouldn't be always the same
						String assignedBrk = "138.37.33.114";
						String assignedPort = "5676";
						props.setProperty(pub_id, assignedBrk);
						FileOutputStream fos =
										 new FileOutputStream(PROPS_FILE_NAME);
						props.storeToXML(fos, PROPS_FILE_DESC);
						//create name of the topic to which the client is listening
						String topicName = PUB_ROOT_TOPIC + pub_id;
						//create publisher for topic
						TopicPublisherAdapter publisher =
								getPublisherForTopic(topicName);
						//create text of the message
						String assignation = assignedBrk + ":" + assignedPort;
						//send the message to the publisher
						sendMessage(publisher, assignation);
						System.out.println("Supercontroller reassigned broker "
								+ assignedBrk + " and port " + assignedPort
								+ " to client " + pub_id);
					}
				}
			}
		} catch (Exception e) {
			e.printStackTrace();
		}

	}
	
	/**
	 * Creates a new supercontroller, which will create all the elements in the
	 * broker failure detection system.
	 * 
	 * @param args  array of arguments from the command line (unused)
	 */
	public static void main(String args[]) {
		new SuperController();
	}

}
